package avicit.bdp.dds.server.worker.task.collect;

import avicit.bdp.common.database.TableMeta;
import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.dto.MetadataStateDto;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.database.adapter.BaseDBAdapter;
import avicit.bdp.common.utils.database.adapter.DbAdapterUtils;
import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.core.enums.MetadataType;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.dispatch.task.collect.MetadataCollectParameters;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.task.AbstractTask;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.bdp.dgs.api.MetadataTableApi;
import avicit.bdp.dgs.metadata.vo.MetadataMessage;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import java.sql.Connection;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import org.apache.shiro.util.CollectionUtils;
import org.slf4j.Logger;

/**
 * @author mayanj 元数据采集任务
 */
public class MetadataCollectTask extends AbstractTask {

  private MetadataCollectParameters metadataCollectParameters;

  private final ProcessService processService;

  /** taskExecutionContext */
  private final TaskExecutionContext taskExecutionContext;

  private final MetadataTableApi metadataTableApi;

  private final RedisCacheHelper redisCacheHelper;

  /**
   * constructor
   *
   * @param taskExecutionContext taskExecutionContext
   * @param logger logger
   */
  public MetadataCollectTask(TaskExecutionContext taskExecutionContext, Logger logger) {
    super(taskExecutionContext, logger);
    this.taskExecutionContext = taskExecutionContext;
    this.processService = SpringApplicationContext.getBean(ProcessService.class);
    this.metadataTableApi = SpringApplicationContext.getBean(MetadataTableApi.class);
    this.redisCacheHelper = SpringApplicationContext.getBean(RedisCacheHelper.class);
  }

  @Override
  public void init() {
    logger.info("metadata collect parameters: {}", taskExecutionContext.getTaskParams());
    metadataCollectParameters =
        JSONUtils.parseObject(
            taskExecutionContext.getTaskParams(), MetadataCollectParameters.class);
    if (metadataCollectParameters == null || !metadataCollectParameters.checkParameters()) {
      throw new RuntimeException("metadata collect parameters is not valid");
    }
  }

  /**
   * run task
   *
   * @throws Exception if error throws Exception
   */
  @Override
  public void handle() throws Exception {
    if (metadataCollectParameters.getMetadataType() != null
        && MetadataType.TABLE.ordinal() == metadataCollectParameters.getMetadataType()) {
      try {
        collectTableMetadata();
        setExitStatusCode(Constants.EXIT_CODE_SUCCESS);
      } catch (Exception e) {
        logger.error("元数据采集异常：", e);
        setExitStatusCode(Constants.EXIT_CODE_FAILURE);
        throw e;
      }
    }
  }

  /**
   * 获取表类型的元数据
   *
   * @throws Exception 获取异常
   */
  public void collectTableMetadata() throws Exception {
    logger.info("开始采集元数据");
    String dataSourceId = metadataCollectParameters.getDataSourceId();
    DataSourceDTO dataSourceDTO = processService.findDataSourceById(dataSourceId);
    if (dataSourceDTO == null) {
      throw new BusinessException("数据源不存在");
    }
    BaseDataSource baseDataSource = DBUtils.getBaseDataSource(dataSourceDTO);

    List<String> tableList;

    try (Connection connection = DBUtils.buildConnectionFromPool(baseDataSource)) {

      BaseDBAdapter dbAdapter = DbAdapterUtils.getDbAdapter(dataSourceDTO.getDatasourceType());

      if (metadataCollectParameters.isAllDatabase()) {
        // 如果是采集全表，首先查询当前数据源中所有的表
        logger.info("全库采集元数据");
        tableList = dbAdapter.getTableList(baseDataSource);
      } else {
        tableList = Arrays.asList(metadataCollectParameters.getTables().split(Constants.COMMA));
      }

      if (CollectionUtils.isEmpty(tableList)) {
        logger.info("待采集表数量为空，结束采集");
        return;
      }

      logger.info("待采集表数量为{}", tableList.size());

      for (String tableName : tableList) {
        logger.info("开始获取表{}的元数据", tableName);
        try {
          // 获取表信息
          TableMeta tableMeta =
              dbAdapter.getDataTable(baseDataSource, connection, tableName, false);
          if (tableMeta != null) {
            logger.info("获取表{}的元数据成功", tableName);

            MetadataStateDto metadataInfo =
                dbAdapter.getMetadataInfo(baseDataSource, connection, tableName, false);
            if (metadataInfo != null) {
              logger.info("获取表{}的技术元数据成功,{}", tableName, JSONUtils.toJsonString(metadataInfo));
              metadataInfo.setDataSourceId(dataSourceId);
              metadataInfo.setDatabaseName(baseDataSource.getDatabase());
              metadataInfo.setDataType(dataSourceDTO.getDatasourceType());
              metadataInfo.setStatDate(new Date());
              tableMeta.setMetadataState(metadataInfo);
            }
            redisCacheHelper.setCacheSet(
                "MetadataCollect_" + taskExecutionContext.getTaskInstanceId(), tableMeta);
          } else {
            logger.error("数据源[{}]表[{}]，没有获取到表信息", dataSourceId, tableName);
          }
        } catch (Exception e) {
          logger.error("数据源[{}]表[{}]，异常", dataSourceId, tableName, e);
        }
      }

      logger.info("元数据采集完毕，将调用接口处理元数据");

      MetadataMessage metadataMessage = new MetadataMessage();
      metadataMessage.setMessageId(ComUtil.getId());
      metadataMessage.setMetadataType(metadataCollectParameters.getMetadataType());
      metadataMessage.setMetadataCollectTaskId(
          metadataCollectParameters.getMetadataCollectTaskId());
      metadataMessage.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());

      boolean handleResult = metadataTableApi.handleMetadataMessage(metadataMessage);
      if (!handleResult) {
        logger.error("元数据采集失败");
      } else {
        logger.info("元数据采集成功");
      }

    } catch (Exception e) {
      logger.error("元数据采集失败", e);
    } finally {
      redisCacheHelper.del("MetadataCollect_" + taskExecutionContext.getTaskInstanceId());
    }
  }

  @Override
  public AbstractParameters getParameters() {
    return metadataCollectParameters;
  }
}
